package cn.ac.iie.zookeeper;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;

/**
 *
 * @author zmc
 */
public class Main {

    static final String zkConnectionString = "192.168.11.94:2181,192.168.11.95:2181,192.168.11.96:2181,192.168.11.97:2181,192.168.11.98:2181";
    static final String dir = "/datadock/executor";
    static final String host = "node1";
    static final String port = "00000";
    static CuratorFramework client = null;

    static {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient(zkConnectionString, retryPolicy);
        client.start();
    }

    public static void main(String[] args) throws Exception {
        //create watch dir
        String path = dir + "/" + host + ":" + port;
        System.out.println("Register node dir:" + path);
        client.create().creatingParentContainersIfNeeded().forPath(path);
        
        watch();
        
        Thread.sleep(Long.MAX_VALUE);
    }

    private static void watch() throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, dir, false);
        cache.start();
        System.out.println("Watcher start ...");
        System.out.println("watch dir:" + dir);

        PathChildrenCacheListener cacheListener = new PathChildrenCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                String path = ZKPaths.getNodeFromPath(event.getData().getPath());
                switch (event.getType()) {
                    case CHILD_ADDED: {
                        System.out.println("Node added:" + path);
                        break;
                    }
                    case CHILD_REMOVED: {
                        System.out.println("Node removed:" + path);
                        break;
                    }
                    case CHILD_UPDATED: {
                        System.out.println("Node updated:" + path);
                        break;
                    }
                }
            }
        };
        //register  listener
        cache.getListenable().addListener(cacheListener);
    }
}
